package cn.me.alphamq.broker.netty.nameserver;

import cn.me.alphamq.broker.cache.Container;
import cn.me.alphamq.broker.config.BrokerProperties;
import cn.me.alphamq.broker.config.ClusterProperties;
import cn.me.alphamq.common.entity.req.PullBrokerAddressReq;
import cn.me.alphamq.common.entity.req.ServiceRegistryReq;
import cn.me.alphamq.common.entity.resp.PullBrokerAddressResp;
import cn.me.alphamq.common.enumeration.BrokerModeEnum;
import cn.me.alphamq.common.enumeration.BrokerRoleEnum;
import cn.me.alphamq.common.enumeration.EventCodeEnum;
import cn.me.alphamq.common.enumeration.ServiceInstanceTypeEnum;
import cn.me.alphamq.common.msg.TcpMsg;
import cn.me.alphamq.common.remote.SyncFuture;
import cn.me.alphamq.common.store.SyncFutureManager;
import cn.me.alphamq.common.util.AssertUtil;
import cn.me.alphamq.common.util.AttrUtil;
import cn.me.alphamq.common.util.NettyUtil;
import com.alibaba.fastjson2.JSON;
import io.netty.channel.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.Inet4Address;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

/**
 * nameserver 的客户端
 *
 * @author f
 */
public class NameServerClient {

    private static final Logger log = LoggerFactory.getLogger(NameServerClient.class);

    /**
     * 与服务端建立连接的 channel
     */
    private Channel channel;

    /**
     * 初始化连接，如果 broker 处于集群模式，则初始化节点之间的连接
     */
    public void initConnection() {
        BrokerProperties brokerProperties = Container.getBrokerProperties();
        ClusterProperties clusterProperties = Container.getClusterProperties();
        channel = NettyUtil.startNettyClient(
                brokerProperties.getNameServerIp(),
                brokerProperties.getNameServerPort(),
                "broker-nameserver-service",
                new BrokerNameServerHandler()
        );

        BrokerModeEnum brokerMode = clusterProperties.getBrokerMode();
        BrokerRoleEnum brokerRole = clusterProperties.getBrokerRole();
        if (BrokerModeEnum.MASTER_SLAVE == brokerMode) {
            if (BrokerRoleEnum.SLAVE == brokerRole) {
                Container.getSlaveSyncClient().initConnection(
                        fetchMasterBrokerAddress(clusterProperties.getBrokerGroup())
                );
            }
        }
    }

    // 获取主 broker 节点的地址
    private String fetchMasterBrokerAddress(String brokerGroup) {
        String msgId = UUID.randomUUID().toString();
        SyncFuture syncFuture = new SyncFuture(msgId);
        SyncFutureManager.put(msgId, syncFuture);
        PullBrokerAddressReq pullBrokerAddressReq = new PullBrokerAddressReq(brokerGroup);
        pullBrokerAddressReq.setMsgId(msgId);
        channel.writeAndFlush(new TcpMsg(EventCodeEnum.PULL_BROKER_ADDRESS_REQ, JSON.toJSONBytes(pullBrokerAddressReq)));
        try {
            return JSON.parseObject(syncFuture.get().getBody(), PullBrokerAddressResp.class).
                    getBrokerAddressList().get(0);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    public Channel getChannel() {
        AssertUtil.isNotNull(channel, "channel has not been connected!");
        return channel;
    }

}
